package com.lmind.core.message;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Queue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.service.IoService;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.lmind.bean.EventProvider;

/**
 * 基于mina2实现的消息管道
 * @author 刘勉
 *
 */
public class MessagePipeIoHandler extends IoHandlerAdapter implements MessageService {

	private static final Logger logger = LoggerFactory.getLogger(MessagePipeIoHandler.class);
	
	private EventProvider<MessageServiceListener> event = new EventProvider<MessageServiceListener>();
	
	private IoService ioService;
	
	private MessageCodec codec;
	
	private AtomicReference<MessagePipeIoHandlerConfig> config = 
			new AtomicReference<MessagePipeIoHandlerConfig>();
	
	public void setIoService(IoService ioService) {
		this.ioService = ioService;
	}

	public void setCodec(MessageCodec codec) {
		this.codec = codec;
	}
	
	public void setConfig(MessagePipeIoHandlerConfig config) {
		this.config.set(config);
	}

	public void init() {
		
		MessageCodecProtocolFactory factory = new MessageCodecProtocolFactory();
		factory.setCodec(codec);
		ioService.getFilterChain().addLast( 
                "codec", new ProtocolCodecFilter(factory)); 
	}

	@Override
	public void sessionCreated(IoSession session) throws Exception {
		
		if (config.get().getTimeout() != 0) {
			session.getConfig().setReaderIdleTime(config.get().getTimeout());
		}
		
		if (config.get().getKeepAliveTime() != 0) {
			session.getConfig().setWriterIdleTime(config.get().getKeepAliveTime());
		}
		
		session.setAttribute(Pipe.class.getName(), new Pipe(session));
	}

	@Override
	public void sessionOpened(IoSession session) throws Exception {
		
		logger.debug("sessionOpened:" + session.getId());
		Pipe pipe = getPipe(session);
		for (MessageServiceListener listener : event.toList()) {
			listener.connected(pipe);
		}
	}

	@Override
	public void sessionClosed(IoSession session) throws Exception {
		Pipe pipe = getPipe(session);
		for (MessageServiceListener listener : event.toList()) {
			listener.disconnected(pipe);
		}
	}

	@Override
	public void sessionIdle(IoSession session, IdleStatus status)
			throws Exception {
		
		if (status.equals(IdleStatus.READER_IDLE)) {
			session.close(true);
			logger.info("会话响应超时:关闭连接" + session.getId());
		} else if (status.equals(IdleStatus.WRITER_IDLE)) {
			Pipe pipe = getPipe(session);
			for (MessageServiceListener listener : event.toList()) {
				listener.keepAlive(pipe);
			}
		}
		
		super.sessionIdle(session, status);
	}

	@Override
	public void exceptionCaught(IoSession session, Throwable cause)
			throws Exception {
		logger.info("exceptionCaught:" + session.getId(), cause);
		session.close(true);
	}

	@Override
	public void messageReceived(IoSession session, Object message)
			throws Exception {
		
		logger.debug("messageReceived:" + message.toString());
		
		Pipe pipe = getPipe(session);
		for (MessageServiceListener listener : event.toList()) {
			listener.message(pipe, message);
		}
	}

	@Override
	public void messageSent(IoSession session, Object message) throws Exception {
		
		logger.debug("messageSent:" + message.toString());
		super.messageSent(session, message);
	}

	@Override
	public void addMessageServiceListener(MessageServiceListener listener) {
		event.addEventListener(listener);
	}

	@Override
	public void removeMessageServiceListener(MessageServiceListener listener) {
		event.removeEventListener(listener);
	}
	
	public List<MessagePipe> getMessagePipes() {
		ArrayList<MessagePipe> list = new ArrayList<MessagePipe>();
		for (IoSession session : ioService.getManagedSessions().values()) {
			MessagePipe pipe = getPipe(session);
			list.add(pipe);
		}
		return list;
	}
	
	public Object getAttribute(long sessionId, Object key) {
		IoSession session = getSession(sessionId);
		return session.getAttribute(key);
	}
	
	public void setAttribute(long sessionId, Object key, Object value) {
		IoSession session = getSession(sessionId);
		session.setAttribute(key, value);
	}
	
	
	protected void send(IoSession session, Object message) {
		session.write(message);
	}
	
	private IoSession getSession(long sessionId) {
		IoSession session = ioService.getManagedSessions().get(sessionId);
		if (session == null) {
			throw new MessagePipeException("该会话不存在或已断开");
		}
		return session;
	}
	
	
	private Pipe getPipe(IoSession session) {
		return (Pipe)session.getAttribute(Pipe.class.getName());
	}
	
	
	class Pipe implements MessagePipe {
		
		private IoSession connection;
		
		private HashMap<Object, Object> attribute = new HashMap<Object, Object>();
		
		public Pipe(IoSession connection) {
			this.connection = connection;
		}

		@Override
		public void send(Object message) {
			connection.write(message);
		}
		
		@Override
		public void close() {
			connection.close(true);
		}

		@Override
		public Object getAttribute(Object key) {
			return attribute.get(key);
		}

		@Override
		public void setAttribute(Object key, Object value) {
			attribute.put(key, value);
		}

		@Override
		public void removeAttribute(Object key) {
			attribute.remove(key);
		}
		
	}

}
